/** To order the required attributes from the result **/
/**
 *  added by akzing(zengzh@comp.nus.edu.sg)
 */
package qp.operators;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.Vector;

import qp.utils.Attribute;
import qp.utils.Batch;
import qp.utils.Schema;
import qp.utils.Tuple;

public class Orderby extends Operator
{

	Operator base;
	Vector<Attribute> attr; // ************************
	Vector<String> AscOrDesc;//**************************
	//boolean asc; // to order by asc or des?
	int batchsize; // number of tuples per outbatch

	int cur; // cursor for the input batch
	boolean eos; // get the end of the object stream
	ObjectInputStream SortedObejctStream;

	/**
	 * The following fields are required during execution of the Orderby
	 * Operator
	 **/

	Batch inbatch;
	Batch outbatch;

	public Orderby(Operator base, Vector<Attribute> as, Vector<String> S, int type)
	{
		super(type);
		this.base = base;
		this.attr = as;//***********
		this.AscOrDesc = S;//*************
		this.cur = 0;
		this.eos = false;
	}

	public void setBase(Operator base)
	{
		this.base = base;
	}

	public Operator getBase()
	{
		return base;
	}

	public Vector<Attribute> getOrderbyAttrlist()//*************
	{
		return attr;
	}

	/**
	 * Opens the connection to the base operator Also figures out what the
	 * tuples to be ordered
	 **/

	public boolean open()
	{
		/** set number of tuples per batch **/
		int tuplesize = schema.getTupleSize();
		batchsize = Batch.getPageSize() / tuplesize;
		Schema baseSchema = base.getSchema();

		if (!base.open())
			return false;
		else
		{
			/**
			 * create temporal file to store the intermediate result. sort the
			 * tuple according to the order by attr
			 */
			String unSortedFile = "OderbyTemp.tem";
			try
			{
				FileOutputStream fos = new FileOutputStream(unSortedFile);
				ObjectOutputStream oos = new ObjectOutputStream(fos);
				Batch tempOutputBatch;

				while ((tempOutputBatch = base.next()) != null)
				{
					oos.writeObject(tempOutputBatch);
				}
				fos.close();
				oos.close();
			} catch (Exception e)
			{
				System.out.println("Orderby Open():write " + unSortedFile + " error");
				return false;
			}
			String sortedFile = "";

			MergeSortByMulAttri es = new MergeSortByMulAttri(unSortedFile, attr, AscOrDesc, base.getSchema());
			try
			{
				sortedFile = es.MergeSort();
			} catch (Exception e)
			{
				System.out.println("Oderby Open():MergeSort error");
				return false;
			}
			
			if (sortedFile != null)
			{
				try
				{
					FileInputStream fileIn = new FileInputStream(sortedFile);
					SortedObejctStream = new ObjectInputStream(fileIn);
				} catch (IOException e)
				{
					System.out.println("Orderby Open():read " + sortedFile + " error");
					return false;
				}
			}
			if (!base.close())
				return false;
			return true;
		}
	}

	/** Read next tuple from operator */

	public Batch next()
	{
		if(eos)
			return null;
		outbatch = new Batch(batchsize);
		while (true)
		{
			if (cur == 0)
			{
				try
				{
					inbatch = (Batch) SortedObejctStream.readObject();
				} catch (IOException io)
				{
					// meet the end of the obejct stream
					eos = true;
					return outbatch;
				} catch (ClassNotFoundException c)
				{
					System.out.println("Orderby next():read SortedObejctStream error");
					System.exit(1);
				}
			}
			for (int i = cur; i < inbatch.size(); i++)
			{
				Tuple oldTuple = inbatch.elementAt(i);
				outbatch.add(oldTuple);
				if (outbatch.isFull())
				{
					cur = i + 1; // remember the position, resume next time
					return outbatch;
				}
			}
			/**
			 * if inbatch runs out, reset the cursor to zero. read the first
			 * object in the next inbatch.
			 */
			cur = 0;
		}
	}

	/** Close the operator */
	public boolean close()
	{
		// File f = new File("OderbyTemp.tem");
		// f.delete();
		try
		{
			SortedObejctStream.close();
		} catch (Exception e)
		{

		}
		return true;
	}

	public Object clone()
	{
		Vector newattri = new Vector();
		for(int i=0;i<attr.size();i++){
		    Attribute newatt = (Attribute) ((Attribute)attr.elementAt(i)).clone();
		    newattri.add(newatt);
		}
		Vector newStri = new Vector();
		for(int i=0;i<AscOrDesc.size();i++){
		    String newatt = AscOrDesc.elementAt(i);
		    newStri.add(newatt);
		}
		Operator newbase = (Operator) base.clone();
		Orderby neworderby = new Orderby(newbase, newattri, newStri, optype);
		Schema newSchema = newbase.getSchema();
		neworderby.setSchema(newSchema);
		return neworderby;
	}
}
